package com.atguigu.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.GmallConfig;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/*
        connectedStream.process(new BroadcastProcessFunction<JSONObject, String, Object>() {
            @Override
            public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<Object> out) throws Exception {

            }

            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<Object> out) throws Exception {

            }
        })
 */
// BroadcastStream<String> broadcastDS = mysqlDS.broadcast(mapStateDescriptor);广播流是String类型
public class DimTableProcessFuntion extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private HashMap<String, TableProcess> tableProcessHashMap;

    //构造方法
    DruidPooledConnection phoenixConn;

    public DimTableProcessFuntion(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //程序运行就先加载mysql的配置信息，以防主流先进processElement方法，配置流还没到
    @Override
    public void open(Configuration parameters) throws Exception {
        tableProcessHashMap = new HashMap<>();
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall-220828-config?" +
                "user=root&password=123456&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false"
        );

        List<TableProcess> tableProcesses = JdbcUtil.queryList(conn,
                "select * from table_process where sink_type='dim'", TableProcess.class, true);

        //将集合数据封装进Map
        for (TableProcess tableProcess : tableProcesses) {
            //校验&建表
            checkTable(tableProcess.getSinkTable(),
                    tableProcess.getSinkColumns(),
                    tableProcess.getSinkPk(),
                    tableProcess.getSinkExtend());

            tableProcessHashMap.put(tableProcess.getSourceTable(), tableProcess);
        }

    }

    /*
            value(flink CDC->广播流->value)，也就是flink CDC输出为：{
    "before":null,
    "after":{
        "source_table":"base_trademark",
        "sink_table":"dim_base_trademark",
        "sink_columns":"id,tm_name",
        "sink_pk":"id",
        "sink_extend":null
    },
    "source":{
        "version":"1.5.4.Final",
        "connector":"mysql",
        "name":"mysql_binlog_source",
        "ts_ms":1655172926148,
        "snapshot":"false",
        "db":"gmall-211227-config",
        "sequence":null,
        "table":"table_process",
        "server_id":0,
        "gtid":null,
        "file":"",
        "pos":0,
        "row":0,
        "thread":null,
        "query":null
    },
    "op":"r",
    "ts_ms":1655172926150,
    "transaction":null
}
             */
    //广播流（配置信息流）
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        /*
        获取广播流状态，把删除的操作产生的value从广播流状态移除，把更新或修改操作产生的value取出after
        数据转化为javabean:TableProcess对象设置到广播流状态的value里，并且要要获取跟phoenix的连接，在phoenix创建好对应的表
         */

        //todo 1.解析数据为javabean：TableProcess对象
        //1.1先把String类型的value转化成jsonObject，这样就可以通过getString获取想要的修改的内容，并且通过JSONObject.parseObject方法把jsonObject转化为javabean了
        JSONObject jsonObject = JSONObject.parseObject(value);//将value转化为json对象//等同于JSON.parseObject，JSONObject是JSON子类


            //todo 1.2 如果当前为删除数据操作，则从状态中也要将数据进行删除（因为processElement方法里主流要根据广播状态过滤自己的数据）
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        if ("d".equals(jsonObject.getString("op"))) {
            String before = jsonObject.getString("before");
            broadcastState.remove(JSON.parseObject(before).getString("source_table"));//remove(key):key是表名，value的after里source_table对应的就是表名
            tableProcessHashMap.remove(JSON.parseObject(before).getString("source_table"));

        } else {

            //不是删除状态的value就取出after数据并转化为TableProcess，用于后面将TableProcess设置到广播状态的value里
            //1.3通过getString获取想要的修改的内容( flink CDC输出里的after里的所有内容，after是修改或增加后的一行数据内容)，并且通过JSONObject.parseObject方法把jsonObject转化为javabean
            TableProcess tableProcess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);

            //todo 2.检验&建表（表有可能之前建过了，所以要检验表是否已经存在）
            if ("dim".equals(tableProcess.getSinkType())) {
                checkTable(tableProcess.getSinkTable(),
                        tableProcess.getSinkColumns(),
                        tableProcess.getSinkPk(),
                        tableProcess.getSinkExtend());

                //todo 3.将数据写入广播状态（因为processElement方法里主流要根据广播状态过滤自己的数据，否则主流获取的广播状态就是空的了）
                broadcastState.put(tableProcess.getSourceTable(), tableProcess);//注意！！！这里的表名一定得是sourceTable,若写的是sinkTable:dim_表名，由于主流没有这个字段，就会导致主流匹配不到这个表名，就会导致什么也匹配不上，数据全丢失了
            }
        }
    }

    //todo 校验&建表
    // sinkColumns:id,tm_name
    //sink_extend:SALT_BUCKETS = 3
    //-> create table if not exists db.sink_table(id varchar primary key,tm_name varchar)SALT_BUCKETS = 3
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        //phoenix建表要指定sinkPK是主键，所以不能为null和"",建表前要判断主键是否为空
        if (sinkPk == null || "".equals(sinkPk)) {
            //如果sinkPk为null，或为""，就设置主键字段名为id，因为一般维度表的主键都是为id
            sinkPk = "id";
        }
        if (sinkExtend == null) {
            sinkExtend = "";
        }

        //拼接建表sql语句
        StringBuilder createSql = new StringBuilder("create table if not exists ")
                .append(GmallConfig.PHOENIX_DB)
                .append(".")
                .append(sinkTable)
                .append("(");
        //切分sinkColumns作为列名并声明主键
        String[] columns = sinkColumns.split(",");

        for (int i = 0; i < columns.length; i++) {
            String column = columns[i];
            //判断是否为主键
            if (sinkPk.equals(column)) {
                createSql.append(sinkPk).append(" varchar primary key");
            } else {
                createSql.append(column).append(" varchar");
            }
            //判断是否为最后一个字段
            if (i == columns.length - 1) {
                createSql.append(")");
            } else {
                createSql.append(",");
            }
        }

        //拼接扩展字段
        createSql.append(sinkExtend);

        //打印数据
        System.out.println("phoenix建表语句为：" + createSql);

        //执行建表sql语句
        //对于异常的处理：建表失败了，继续往下走，也无法写入phoenix，干脆到这里不让它往下走了,所以try catch一下
        PreparedStatement preparedStatement = null;
        try {
            //ctrl+alt+t
            phoenixConn = DruidDSUtil.getPhoenixConn();
            preparedStatement = phoenixConn.prepareStatement(createSql.toString());
            preparedStatement.execute();
        } catch (SQLException e) {
//            e.printStackTrace();
            throw new RuntimeException("创建表：" + sinkTable + "失败！！！！！！");
        }


        //关闭连接
        try {
            preparedStatement.close();
            phoenixConn.close();

        } catch (SQLException e) {
            throw new RuntimeException();
        }
        //如果写在open方法里，该连接就不能关闭，因为open只调用一次，关掉以后连接没有了就无法建表了


    }

    // 主流，从kafka topic_db消费的数据,kafka数据来源于maxwell读取mysql的数据
    /*
    Maxwell数据格式，也就是value:
{"database":"gmall","table":"cart_info","type":"update","ts":1592270938,"xid":13090,"xoffset":1573,"data":{"id":100924,"user_id":"93","sku_id":16,"cart_price":4488,"sku_num":1,"img_url":"http://47.93.148.192:8080/group1/M0rBHu8l-sklaALrngAAHGDqdpFtU741.jpg","sku_name":"华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万30倍数字变焦 8GB+128GB亮黑色全网通5G手机","is_checked":null,"create_time":"2020-06-14 09:28:57","operate_time":null,"is_ordered":1,"order_time":"2021-10-17 09:28:58","source_type":"2401","source_id":null},"old":{"is_ordered":0,"order_time":null}}

{
    "database":"gmall","table":"cart_info","type":"update","ts":1592270938,"xid":13090,"xoffset":1573,
    "data":{//最新字段
    "id":100924,"user_id":"93","sku_id":16,"cart_price":4488,"sku_num":1
    },
    "old":{
        "is_ordered":0,"order_time":null
    }
}
     */
//    processElement方法里主流要根据广播状态过滤自己的数据
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //todo 1.读取数据状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据主流的表名作为key获取广播流的value，有可能获取的是null，因为配置流很有可能没有这个表名
        String keyTableName = value.getString("table");
        TableProcess tableProcess = broadcastState.get(keyTableName);
        TableProcess tableProcessFromMap = tableProcessHashMap.get(keyTableName);

        //todo 2.过滤主流的行和列（根据选取需要的字段：要跟sinkColumns保持一致）
        //2.过滤列：根据Phoenix字段（跟sinkColumns里的字段一致：需要的字段，主流不需要的字段也会过来，需要过滤）过滤数据（行、列）
        String typeValue = value.getString("type");
        if ((tableProcess != null || tableProcessFromMap != null) && ("insert".equals(typeValue) || "update".equals(typeValue) || "bootstrap-insert".equals(typeValue))) {//如果gmall数据库里的表内容是删除操作（配置表删除一行数据），可以选择phoenix不删除表和内容
            //"insert" "update" "bootstrap-insert"是gmall数据库的表的操作，如果有修改，maxwell会监控到生成json字符串，放到kafka里，主流通过flink CDC从kafka里获取这些json字符串

            //todo 数据过滤 过滤行和列
            //value.getJSONObject("data")取的是地址值，filterColumns返回的是void，没有返回值，由于value.get取的是地址值，地址值不变，所以是直接对value数据作了过滤
            if (tableProcessFromMap != null) {
                tableProcess = tableProcessFromMap;
            }
            filterColumns(value.getJSONObject("data"), tableProcess.getSinkColumns());//获取data数据还要加工，所以用getJSONObject

            //todo 3.将过滤后的数据输出
            //补充写到phoenix的表名
            value.put("sink_table", tableProcess.getSinkTable());//************************?************************************

            out.collect(value);//此时的value是过滤后的value，之所以收集value，是因为我们后面还需要value的其他信息，例如要把里面的表名取出来写到phoenix
        } else if ("delete".equals(typeValue) || "bootstrap-start".equals(typeValue) || "completed".equals(typeValue)) {
            System.out.println("主流表是删除操作或bootstrap-start或completed");
        } else if(tableProcess == null){
            //tableProcess == null
            System.out.println("主流表" + keyTableName + "不在配置表信息里！");

        }


    }

    /**
     * @param data        {"id":"1001","tm_name":"TCL","logo_url":"xxx"}
     * @param sinkColumns 1001logo_url,TCL,
     */
    //todo 过滤列方法：根据Phoenix字段（跟sinkColumns里的字段一致：需要的字段，主流不需要的字段也会过来，需要过滤）
    private void filterColumns(JSONObject data, String sinkColumns) {
        String[] columns = sinkColumns.split(",");//columns没有.contains方法，要转化为list，list有这个方法
        List<String> columnsList = (List<String>) Arrays.asList(columns);

        //写法一：
//        Set<Map.Entry<String, Object>> entries = data.entrySet();
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//
//        while (iterator.hasNext()){
//            Map.Entry<String, Object> next = iterator.next();
//            //错误写法：1001logo_url包含logo_url，则logo_url这个字段就不会被移除掉
////            if(!sinkColumns.contains(next.getKey())){//字符串之间会有包含关系
////                iterator.remove();
////            }
//            //正确写法：
//                if(!columnsList.contains(next.getKey())){//这个就表示list集合里有没有主流的那个列名
//                iterator.remove();
//            }

        //写法二：
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnsList.contains(next.getKey()));


    }


}




